package cn.study.generate2;

import cn.study.generate1.Trade;
import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {  
    public static void main(String[] args) throws InterruptedException {  
       
    	long beginTime=System.currentTimeMillis();  
        int bufferSize=1024;  
        ExecutorService executor=Executors.newFixedThreadPool(8);  

        Disruptor<Trade> disruptor = new Disruptor<Trade>(new EventFactory<Trade>() {
            @Override  
            public Trade newInstance() {  
                return new Trade();  
            }  
        }, bufferSize, executor, ProducerType.SINGLE, new BusySpinWaitStrategy());  


        /**菱形操作
         * 主要实现逻辑
         * 一个生产者P1，3个消费者C1、C2、C3
         * P1生产一个数据 A，然后disruptor将数据 A 同时并行的流入到C1、C2的处理器中
         * 在C1、C2处理完数据A后，再将数据A流入到给C3的处理器中
         * 从头到尾所有的消费者处理器都只处理一个数据A
         */
        // 在现实世界中的对应的案例就像：
        // 在处理实际的业务逻辑（C3）之前，需要校验数据（C1），以及将数据写入磁盘（C2）。

//        //使用disruptor创建消费者组C1、C2 ,连接多个消费事件
//        EventHandlerGroup<Trade> handlerGroup =
//        		disruptor.handleEventsWith(new Handler1(), new Handler2());
//        //声明在C1,C2完事之后执行JMS消息发送操作 也就是流程走到C3
//        handlerGroup.then(new Handler3());

        /**
         * 顺序操作 p1--->h1--->h2--->h3
         */
//        disruptor.handleEventsWith(new Handler1()).
//        	handleEventsWith(new Handler2()).
//        	handleEventsWith(new Handler3());


        //六边形操作.
        /**
         * 执行步骤
         * 1. p1--->h1、h2 (并行流入h1、h2)
         * 2. h1--->h4(先执行h1)、h2--->h5 (先执行h2){整体是并行的}
         * 3. h4、h5--->h3(并行流入h3)
         */
//        Handler1 h1 = new Handler1();
//        Handler2 h2 = new Handler2();
//        Handler3 h3 = new Handler3();
//        Handler4 h4 = new Handler4();
//        Handler5 h5 = new Handler5();
//        disruptor.handleEventsWith(h1, h2);
//        disruptor.after(h1).handleEventsWith(h4);
//        disruptor.after(h2).handleEventsWith(h5);
//        disruptor.after(h4, h5).handleEventsWith(h3);

        
        
        disruptor.start();//启动  
        CountDownLatch latch=new CountDownLatch(1);  
        //生产者准备  (创建新的线程TraderPublisher生产数据，并发布)
        executor.submit(new TradePublisher(latch, disruptor));

        //当TradePublisher(生产者)中生产完后(执行了CountDownLatch.latch.countDown();)
        latch.await();//等待生产者完事. 

        //当TradePublisher(生产者)中生产完后(执行了CountDownLatch.latch.countDown();)后，
        //继续执行主线程,关闭disruptor和线程池
        disruptor.shutdown();  
        executor.shutdown();  
        System.out.println("总耗时:"+(System.currentTimeMillis()-beginTime));  
    }  
}  